package drds.data_propagate.parse;

import com.lmax.disruptor.*;
import drds.data_propagate.binlog.BinLogContext;
import drds.data_propagate.binlog.BinLogEvent;
import drds.data_propagate.binlog.Buffer;
import drds.data_propagate.binlog.event.FormatDescriptionEvent;
import drds.data_propagate.common.AbstractLifeCycle;
import drds.data_propagate.common.utils.NamedThreadFactory;
import drds.data_propagate.driver.packets.GtidSet;
import drds.data_propagate.parse.exception.ParseException;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;

/**
 * 针对解析器提供一个多阶段协同的处理
 *
 * <pre>
 * 1. 网络接收 (单线程)
 * 2. 事件基本解析 (单线程，事件类型、DDL解析构造TableMeta、维护位点信息)
 * 3. 事件深度解析 (多线程, DML事件数据的完整解析)
 * 4. 投递到store (单线程)
 * </pre>
 *
 * @author agapple 2018年7月3日 下午4:54:17
 * @since 1.0.26
 */
public class MultiStageCoprocessorImpl extends AbstractLifeCycle implements MultiStageCoprocessor {

    private static final int maxFullTimes = 10;
    public LogEventConvert logEventConvert;
    public EventList eventList;
    public Connection connection;
    public volatile ParseException exception;
    public GtidSet gtidSet;
    private int parserThreadCount;
    private int ringBufferSize;
    private RingBuffer<MessageEvent> messageEventRingBuffer;
    private ExecutorService parserExecutor;
    private ExecutorService stageExecutor;
    private String destination;
    private AtomicLong eventsPublishBlockingTime;
    private WorkerPool<MessageEvent> workerPool;
    private BatchEventProcessor<MessageEvent> simpleParserStage;
    private BatchEventProcessor<MessageEvent> sinkStoreStage;
    private BinLogContext binLogContext;

    public MultiStageCoprocessorImpl(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert,
                                     EventList eventList, String destination) {
        this.ringBufferSize = ringBufferSize;
        this.parserThreadCount = parserThreadCount;
        this.logEventConvert = logEventConvert;
        this.eventList = eventList;
        this.destination = destination;
    }

    @Override
    public void start() {
        super.start();
        this.exception = null;
        this.messageEventRingBuffer = RingBuffer.createSingleProducer(new MessageEventFactory(), ringBufferSize, new BlockingWaitStrategy());
        int tc = parserThreadCount > 0 ? parserThreadCount : 1;
        this.parserExecutor = Executors.newFixedThreadPool(tc, new NamedThreadFactory("MultiStageCoprocessor-Parser" , destination));

        this.stageExecutor = Executors.newFixedThreadPool(2, new NamedThreadFactory("MultiStageCoprocessor-other", destination));
        SequenceBarrier sequenceBarrier = messageEventRingBuffer.newBarrier();
        ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler();
        // stage 2
        this.binLogContext = new BinLogContext();
        simpleParserStage = new BatchEventProcessor<MessageEvent>(messageEventRingBuffer,
                sequenceBarrier,
                new ParserStage(this, binLogContext));
        simpleParserStage.setExceptionHandler(exceptionHandler);
        messageEventRingBuffer.addGatingSequences(simpleParserStage.getSequence());

        // stage 3
        SequenceBarrier dmlParserSequenceBarrier = messageEventRingBuffer.newBarrier(simpleParserStage.getSequence());
        WorkHandler<MessageEvent>[] workHandlers = new DmlParserStage[tc];
        for (int i = 0; i < tc; i++) {
            workHandlers[i] = new DmlParserStage(this);
        }
        workerPool = new WorkerPool<MessageEvent>(messageEventRingBuffer,
                dmlParserSequenceBarrier,
                exceptionHandler,
                workHandlers);
        Sequence[] sequence = workerPool.getWorkerSequences();
        messageEventRingBuffer.addGatingSequences(sequence);

        // stage 4
        SequenceBarrier sinkSequenceBarrier = messageEventRingBuffer.newBarrier(sequence);
        sinkStoreStage = new BatchEventProcessor<MessageEvent>(messageEventRingBuffer,
                sinkSequenceBarrier,
                new SinkStoreStage(this));
        sinkStoreStage.setExceptionHandler(exceptionHandler);
        messageEventRingBuffer.addGatingSequences(sinkStoreStage.getSequence());

        // start work
        stageExecutor.submit(simpleParserStage);
        stageExecutor.submit(sinkStoreStage);
        workerPool.start(parserExecutor);
    }

    public void setBinlogChecksum(int binlogChecksum) {
        if (binlogChecksum != BinLogEvent.BINLOG_CHECKSUM_ALG_OFF) {
            binLogContext.setFormatDescriptionEvent(new FormatDescriptionEvent(4, binlogChecksum));
        }
    }

    @Override
    public void stop() {
        // fix bug #968，对于pool与
        workerPool.halt();
        simpleParserStage.halt();
        sinkStoreStage.halt();
        try {
            parserExecutor.shutdownNow();
            while (!parserExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
                if (parserExecutor.isShutdown() || parserExecutor.isTerminated()) {
                    break;
                }

                parserExecutor.shutdownNow();
            }
        } catch (Throwable e) {
            // ignore
        }

        try {
            stageExecutor.shutdownNow();
            while (!stageExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
                if (stageExecutor.isShutdown() || stageExecutor.isTerminated()) {
                    break;
                }

                stageExecutor.shutdownNow();
            }
        } catch (Throwable e) {
            // ignore
        }
        super.stop();
    }

    public boolean publish(Buffer buffer) {
        return this.publish(buffer, null);
    }

    /**
     * 网络数据投递
     */
    public boolean publish(BinLogEvent binLogEvent) {
        return this.publish(null, binLogEvent);
    }

    private boolean publish(Buffer buffer, BinLogEvent binLogEvent) {
        if (!isStart()) {
            if (exception != null) {
                throw exception;
            }
            return false;
        }

        /**
         * 由于改为processor仅终止自身stage而不是stop，那么需要由incident标识coprocessor是否正常工作。
         * 让dump线程能够及时感知
         */
        if (exception != null) {
            throw exception;
        }
        boolean interupted = false;
        long blockingStart = 0L;
        int fullTimes = 0;
        do {
            try {
                long nextIndex = messageEventRingBuffer.tryNext();
                MessageEvent messageEvent = messageEventRingBuffer.get(nextIndex);
                if (buffer != null) {
                    messageEvent.setBuffer(buffer);
                } else {
                    messageEvent.setBinLogEvent(binLogEvent);
                }
                messageEventRingBuffer.publish(nextIndex);
                if (fullTimes > 0) {
                    eventsPublishBlockingTime.addAndGet(System.nanoTime() - blockingStart);
                }
                break;
            } catch (InsufficientCapacityException e) {
                if (fullTimes == 0) {
                    blockingStart = System.nanoTime();
                }
                // park
                // LockSupport.parkNanos(1L);
                applyWait(++fullTimes);
                interupted = Thread.interrupted();
                if (fullTimes % 1000 == 0) {
                    long nextStart = System.nanoTime();
                    eventsPublishBlockingTime.addAndGet(nextStart - blockingStart);
                    blockingStart = nextStart;
                }
            }
        } while (!interupted && isStart());
        return isStart();
    }

    // 处理无数据的情况，避免空循环挂死
    private void applyWait(int fullTimes) {
        int newFullTimes = fullTimes > maxFullTimes ? maxFullTimes : fullTimes;
        if (fullTimes <= 3) { // 3次以内
            Thread.yield();
        } else { // 超过3次，最多只sleep 1ms
            LockSupport.parkNanos(100 * 1000L * newFullTimes);
        }

    }

    public void setLogEventConvert(LogEventConvert logEventConvert) {
        this.logEventConvert = logEventConvert;
    }

    public void setEventList(EventList eventList) {
        this.eventList = eventList;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    public void setEventsPublishBlockingTime(AtomicLong eventsPublishBlockingTime) {
        this.eventsPublishBlockingTime = eventsPublishBlockingTime;
    }

    public void setGtidSet(GtidSet gtidSet) {
        this.gtidSet = gtidSet;
    }

}
